package com.example.kafkademo.interceptor;

import org.apache.kafka.clients.producer.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

/**
 * @author zyy
 * @version 1.0
 * @date 2020/10/20 17:28
 *
 */
public class KafkaInterceptor {

    public static void main(String[] args){
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "10.42.25.12:9092,10.42.25.13:9092,10.42.25.15:9092,");
        properties.put("batch.size", 65536);
        properties.put("linger.ms", 10);
        properties.put("buffer.memory", 134217728);
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 开启GZIP压缩
        properties.put("compression.type", "zstd");   //Producer 端压缩、Broker 端保持、Consumer 端解压缩。
        properties.put("acks", "all");   //表明所有副本 Broker 都要接收到消息，该消息才算是“已提交”
        // TODO  消息重试 和 callback 有什么区别？     个人理解重试可以处理网络抖动， callback 可以处理其他异常
        properties.put("retries", "3");  // kafka 消息重试次数

        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.example.kafkademo.interceptor.AvgLatencyProducerInterceptor"); // 拦截器 1
//        interceptors.add("com.example.kafkademo.interceptor.AvgLatencyConsumerInterceptor"); // 拦截器 2
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);

        /**
         * ry-with-resource 特性，所以并没有显式调用 producer.close() 方法。
         */
        try (Producer<String, String> producer = new KafkaProducer<String, String>(properties)){
            for (int i = 0; i < 10; i++) {
                // TODO 日志貌似创建时只会连接一个节点
                ProducerRecord<String, String> record = new ProducerRecord<>("test-a", String.valueOf(i));
                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if(e == null){
                            // No Exception
                        }else{
                            //Exception Handling
                        }
                    }
                });

            }

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
